package com.sam.kafka.consumer;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;

import java.util.*;

/**
 * @author: wm.xue
 * @date: 2021/11/9 18:06
 * @description: 自定义消费者位移
 */
public class _04_Consumer_MyOffset {
    public static void main(String[] args) {

        Map<String, Object> configs = new HashMap<>();

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(configs);

        //  给当前消费者手动分配一系列主题分区 不能与subscribe一起使用
        consumer.assign(Arrays.asList(new TopicPartition("topic", 0)));

        // 获取给当前消费者分配的分区集合
        Set<TopicPartition> assignments = consumer.assignment();
        for (TopicPartition assignment : assignments) {
            System.err.println("topic：" + assignment.topic() + "\t" +
                    "partition：" + assignment.partition());
        }

        // 获取当前消费者所有主题分区元数据
        Map<String, List<PartitionInfo>> listTopics = consumer.listTopics();
        listTopics.forEach((k,v)->{
            System.err.println("主题：" + k);
            for (PartitionInfo partitionInfo : v) {
                System.err.println("topic：" +partitionInfo.topic() + "\t" +
                        "partition：" + partitionInfo.partition());
            }
        });

        // 获取指定分区的元数据
        List<PartitionInfo> topic = consumer.partitionsFor("topic");

        // 获取主题分区，列出它们第一个消息的偏移量。如果指定的分区不存在，该方法可能会永远阻塞
        Map<TopicPartition, Long> topic1 = consumer.beginningOffsets(Arrays.asList(new TopicPartition("topic", 0)));
        topic1.forEach((k,v)->{
            System.err.println("主题：" + k.topic() + "\t分区：" + k.partition() + "偏移量\t" + v);
        });

        // 寻找最后一个稳定的偏移量，，即下一个要消费的消息现在还是未提交状态的事务消息
        consumer.seekToEnd(consumer.assignment());

        // 将分区的消费者偏移量移动到他们的起始偏移量，
        consumer.seekToBeginning(consumer.assignment());

        // 将指定分区主题的消费偏移量移动到指定的偏移量，即当前消费者下一条要消费的消息偏移量，消费中随意调用，可能会造成数据丢失
        consumer.seek(new TopicPartition("TOPIC", 0), 10);

        // 检查指定主题分区的消费偏移量
        consumer.position(new TopicPartition("TOPIC", 0));

    }
}
